/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.wordcount;

import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;

import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;
import cgl.narada.event.NBEvent;

/**
 * An example demonstrating the <code>Reduce</code> role within a Word Count
 * operation.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class WordCountReduce extends MapReduceBase {
    private Hashtable<String, Integer> wordCounts;


    public WordCountReduce() {
        setDomain("Generic");
        setVersionInformation(200);
        setOperationIdentifier("WordCount-ReduceFunction-GranulesSample");
        wordCounts = new Hashtable<String, Integer>();
    }


    public void execute() {

        try {
            //System.out.println("WordCountReduce is ...... ");
            StreamingAccess mapReduceDataset = getMapReduceDataset();

            while (mapReduceDataset.isDataAvailable()) {
                //System.out.println("executing!");
                NBEvent nbEvent = mapReduceDataset.getStreamingData();
                processResultsFromMap(nbEvent);
            }

        } catch (DatasetException e) {
            e.printStackTrace();
        } catch (IOException e) {
            System.out.println("Problems converting bytes to longs");
            e.printStackTrace();
        } catch (MarshallingException e) {
            System.out.println("Problems Marshalling: " + e.getMessage());
            e.printStackTrace();
        } catch (MapReduceException e) {
            System.out.println("Problems writing Results: " + e.getMessage());
            e.printStackTrace();
        }
    }


    private synchronized void processResultsFromMap(NBEvent nbEvent) throws IOException,
        MarshallingException, MapReduceException {

        String synopsis = (String) nbEvent.getContentSynopsis();
        byte[] resultBytes = nbEvent.getContentPayload();
        Results results = ResultsFactory.getInstance().getResults(resultBytes);
        byte[] reportBytes = results.getResultPayload();

        ResultPayload resultPayload = ResultPayload.getInstance();
        Hashtable mapCounts = resultPayload.getHashtable(reportBytes);

        Enumeration tokens = mapCounts.keys();
        while (tokens.hasMoreElements()) {
            String token = (String) tokens.nextElement();
            int numOfTimes = ((Integer) mapCounts.get(token)).intValue();

            if (!wordCounts.containsKey(token)) {
                wordCounts.put(token, 0);
            }

            int currentCount = wordCounts.get(token);
            currentCount += numOfTimes;
            wordCounts.put(token, currentCount);
        }

        setReceivedOutputFrom(results.getInstanceIdentifier());
        //System.out.println("Received results on [" + synopsis + "] from MAP ("
        //    + results.getInstanceIdentifier() + ")");
        if (!isAwaitingOutputs()) {
        	setTeminationConditionReached();
            System.out.println(synopsis + ":-> " + wordCounts);            
        }
        
    }

}
